package com.noahsloan.nutils.streams.clocked;

/**
 * {@link RateCalculator} that takes measurements over fixed intervals. The last
 * complete interval measurement is always the one returned.
 * 
 * @author noah
 * 
 */
public class FixedWindowCalc implements Clocked, RateCalculator {

    private final int intervalTicks;

    private final TimeSource source;

    private volatile long currentCount;

    private volatile double lastReading;

    private volatile int tickCount;

    private volatile boolean closed;

    /**
     * The start of the time interval we are currently measuring.
     */
    private volatile long intervalStart;

    /**
     * 
     * @param interval
     *            the length of the interval to measure
     * @param source
     *            the time source
     */
    public FixedWindowCalc(long interval, TimeSource source) {
        intervalTicks = (int) Math.ceil((double) interval / source.getTick());
        this.source = source;
        intervalStart = source.currentTime();
        // register to receiver hearbeats
        source.register(this);
    }

    public synchronized void heartbeat() {
        if (!closed) {
            tickCount++;
            tickCount %= intervalTicks;
            if (tickCount == 0) {
                long time = source.currentTime();
                // since we have no real time guarantees, make sure we average
                // over the correct amount of time, even though the intervals
                // should be fixed
                lastReading = currentCount / (time - intervalStart);
                intervalStart = time;
                currentCount = 0;
            }
        }
    }

    public synchronized void close() {
        closed = true;
        source.unregister(this);
    }

    public double getRate() {
        return lastReading;
    }

    public synchronized long storeMeasurement(final long count) {
        currentCount += count;
        return count;
    }

    @Override
    protected void finalize() throws Throwable {
        close();
        super.finalize();
    }
}
